# 调用缓存存储RDD
#coding:utf8
import time

from pyspark import SparkContext,SparkConf
from pyspark.storagelevel import StorageLevel

if __name__ == '__main__':
    conf = SparkConf().setAppName('test_persit').setMaster("local[*]")
    sc = SparkContext(conf=conf)

    rdd = sc.textFile('../data/input/words.txt')

    # 如果没有做缓存，这里对rdd2进行收集后就会被清除
    rdd_split = rdd.map(lambda x: x.split(' '))

    # 以下两个代码一样的效果，都是将RDD存储到缓存中
    # rdd_split = rdd.cache()
    # rdd_split = rdd.persist()

    # 有多种策略可选，这里选择在磁盘和内存中都存放
    rdd_split = rdd_split.persist(StorageLevel.MEMORY_AND_DISK_2)

    print(rdd_split.flatMap(lambda x:x).map(lambda x:(x,1)).reduceByKey(lambda x,y:x+y).collect())

    # 如果没有缓存，这里会从sc.textFile重新加载RDD
    print(rdd_split.flatMap(lambda x:x).map(lambda x:(x,1)).reduceByKey(lambda x,y:x+y).collect())

    # 注意释放缓存
    rdd_split.unpersist()
    time.sleep(49999)